package org.atguigu.gmall.realtime.app.dwd.db;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.atguigu.gmall.realtime.app.BaseApp;
import org.atguigu.gmall.realtime.bean.TableProcess;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.atguigu.gmall.realtime.util.JdbcUtil;

import java.sql.Connection;
import java.util.*;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dwd.db
 * Author: ZengHaiFeng
 * CreateTime:2023/4/26 19:42:41
 * Description: Todo
 * Version：1.0
 */
@Slf4j
public class Dwd_09_BaseDb extends BaseApp {
    public static void main(String[] args) {
        new Dwd_09_BaseDb().init(30009,2, Constant.TOPIC_ODS_DB,"Dwd_09_BaseDb");
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> ds) {
        SingleOutputStreamOperator<JSONObject> etl = etl(ds);
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect = connect(etl, tpStream);
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = deleteNotNeedColumns(connect);
        writeToKafka(resultStream);
    }

    private void writeToKafka(
            SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        resultStream.sinkTo(FlinkSinkUtil.getKafkaSink());
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> deleteNotNeedColumns(
            SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect) {
        return connect.map(t -> {
            JSONObject data = t.f0;
            ArrayList<String> columns = new ArrayList<>(Arrays.asList(t.f1.getSinkColumns().split(",")));
            columns.add("op_type");
            data.keySet().removeIf(key -> !columns.contains(key));
            return t;
        }).returns(Types.TUPLE(Types.GENERIC(JSONObject.class),Types.POJO(TableProcess.class)));
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(
            SingleOutputStreamOperator<JSONObject> etl,
            SingleOutputStreamOperator<TableProcess> tpStream) {
        MapStateDescriptor<String, TableProcess> tpState = new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpState);
        return etl.connect(tpBcStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject,TableProcess>>() {
                    private final HashMap<String,TableProcess> tpMap = new HashMap<>();
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 确保配置信息比数据先到
                        // 在 open method中先加载全部配置： 预加载技术 来解决 动态分流技术中 数据比广播的配置信息先到的情况
                        // connect mysql
                        Connection mysqlConnection = JdbcUtil.getMysqlConnection();
                        // query all of the configurations on mysql table
                        String sql = "select * from gmall2023_config.table_process";
                        List<TableProcess> tps = JdbcUtil.queryList(mysqlConnection, sql, null, TableProcess.class, true);
                        mysqlConnection.close();
                        for (TableProcess tp : tps) {
                            tpMap.put(getKey(tp.getSourceTable(),tp.getSourceType()),tp);
                        }
                    }

                    private String getKey(String sourceTable, String sourceType) {
                        return sourceTable + ":" + sourceType;
                    }

                    @Override
                    public void processElement(
                            JSONObject value,
                            ReadOnlyContext ctx,
                            Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        String table = value.getString("table");
                        String type = value.getString("type");
                        String key = getKey(table, type);
                        ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpState);
                        TableProcess tp = state.get(key);
                        if (tp == null){
                            // 状态中没有找到配置信息 就去map中找
                            tp = tpMap.get(key);
                        }
                        if (tp != null){
                            // 输出 data
                            JSONObject data = value.getJSONObject("data");
                            data.put("op_type",value.getString("type"));
                            out.collect(Tuple2.of(data,tp));
                        }
                    }

                    @Override
                    public void processBroadcastElement(
                            TableProcess value,
                           Context ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpState);

                        String key = getKey(value.getSourceTable(), value.getSourceType());
                        if ("d".equals(value.getOp())){
                            state.remove(key);
                            tpMap.remove(key);
                        }else {
                            state.put(key,value);
                        }
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(
            StreamExecutionEnvironment env) {
        Properties props = new Properties();
        props.setProperty("useSSL","false");
        MySqlSource<String> mysql = MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_PORT)
                .databaseList("gmall2023_config")
                .tableList("gmall2023_config.table_process")
                .username("root")
                .password("aaaaaa")
                .jdbcProperties(props)
                .serverTimeZone("Asia/Shanghai")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
        return env.fromSource(mysql, WatermarkStrategy.noWatermarks(),"mysql")
                .map(value -> {
                    JSONObject obj = JSON.parseObject(value);
                    String op = obj.getString("op");
                    TableProcess tp;
                    if ("d".equals(op)){
                        tp = obj.getObject("before", TableProcess.class);
                    }else {
                        tp = obj.getObject("after", TableProcess.class);
                    }
                    tp.setOp(op);
                    return tp;
                })
                .filter(v -> "dwd".equals(v.getSinkType()));
    }

    private SingleOutputStreamOperator<JSONObject> etl(
            DataStreamSource<String> ds) {
        return ds.filter(value -> {
            try {
                JSONObject obj = JSON.parseObject(value);
                String type = obj.getString("type");
                return obj.getString("database").equals("gmall2023")
                        && (type.equals("insert") || type.equals("update")
                            || type.equals("delete") || type.equals("bootstrap-insert"))
                        && null != obj.getJSONObject("data");
            }catch (Exception e){
                log.warn("not json" + value);
                return false;
            }
        }).map(json -> JSON.parseObject(json.replaceAll("bootstrap-","")));
    }
}
